﻿using System;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using System.Collections;
using System.Collections.Concurrent;
using System.Collections.Generic;
using Orleans.Runtime;
using System.Linq;
using Orleans;

namespace IOP.Orleans
{

    /// <summary>
    /// 观察者管理器
    /// </summary>
    /// <typeparam name="TObserver"></typeparam>
    public class ObserverManager<TObserver> : IObserverManager<TObserver>
        where TObserver : class, IAddressable
    {
        /// <summary>
        /// 日志前缀
        /// </summary>
        private readonly string _LogPrefix;
        /// <summary>
        /// 观察者字典
        /// </summary>
        private readonly ConcurrentDictionary<IAddressable, TObserver> _Observers = new ConcurrentDictionary<IAddressable, TObserver>();
        /// <summary>
        /// 日志
        /// </summary>
        private readonly ILogger _Logger;

        /// <summary>
        /// 构造函数
        /// </summary>
        /// <param name="expiration">超时时间</param>
        /// <param name="log">日志</param>
        /// <param name="logPrefix">日志前缀</param>
        public ObserverManager(ILogger log, string logPrefix)
        {
            _Logger = log;
            _LogPrefix = logPrefix;
            GetDateTime = () => DateTime.UtcNow;
        }

        /// <summary>
        /// 当前时间
        /// </summary>
        public Func<DateTime> GetDateTime { get; set; }
        /// <summary>
        /// 订阅者数量
        /// </summary>
        public int Count => _Observers.Count;
        /// <summary>
        /// 获取当前观察者的拷贝
        /// </summary>
        public IDictionary<IAddressable, TObserver> Observers => _Observers.ToDictionary(_ => _.Key, _ => _.Value);
        /// <summary>
        /// 判断一个监听者是否已经订阅了消息
        /// </summary>
        /// <param name="observer"></param>
        /// <returns></returns>
        public bool IsSubscribed(TObserver observer) => Observers.Values.Contains(observer);
        /// <summary>
        /// 判断一个监听者是否已经订阅了消息
        /// </summary>
        /// <param name="observer"></param>
        /// <returns></returns>
        public bool IsSubscribed(IGrainObserver observer)
        {
            TObserver addressable = observer as TObserver;
            return Observers.Values.Contains(addressable);
        }
        /// <summary>
        /// 移除所有观察者
        /// </summary>
        public void Clear() => _Observers.Clear();

        /// <summary>
        /// 订阅服务
        /// </summary>
        /// <param name="address"></param>
        /// <param name="observer"></param>
        public void Subscribe(TObserver observer)
        {
            TObserver entry;
            if (_Observers.TryGetValue(observer, out entry))
            {
                entry = observer;
                if (_Logger.IsEnabled(LogLevel.Debug))
                    _Logger.LogDebug(_LogPrefix + ": Updating entry for {0}/{1}. {2} total subscribers.", observer, observer, _Observers.Count);
            }
            else
            {
                _Observers.AddOrUpdate(observer, observer, (key, value) => value);
                if (_Logger.IsEnabled(LogLevel.Debug))
                    _Logger.LogDebug(_LogPrefix + ": Adding entry for {0}/{1}. {2} total subscribers after add.", observer, observer, _Observers.Count);
            }
        }
        /// <summary>
        /// 取消订阅
        /// </summary>
        /// <param name="subscriber"></param>
        public void Unsubscribe(TObserver subscriber)
        {
            _Logger.LogDebug(_LogPrefix + ": Removed entry for {0}. {1} total subscribers after remove.", subscriber, _Observers.Count);
            _Observers.TryRemove(subscriber, out _);
        }

        /// <summary>
        /// 向所有的客户端发送通知
        /// </summary>
        /// <param name="notification"></param>
        /// <param name="predicate"></param>
        /// <returns></returns>
        public async Task Notify(Func<TObserver, Task> notification, IEnumerable<TObserver> targetObserver = null)
        {
            var defunct = default(List<IAddressable>);
            foreach (var observer in Observers)
            {
                //过滤客户端
                if (targetObserver != null && !targetObserver.Contains(observer.Value)) continue;
                try
                {
                    await notification(observer.Value);
                }
                catch (Exception e)
                {
                    _Logger.LogError(e.Message + "\r\n" + e.StackTrace);
                    defunct ??= new List<IAddressable>();
                    defunct.Add(observer.Key);
                }
            }

            //移除出问题的客户端
            if (defunct != default(List<IAddressable>))
            {
                foreach (var observer in defunct)
                {
                    _Observers.TryRemove(observer, out _);
                    if (_Logger.IsEnabled(LogLevel.Debug))
                        _Logger.LogDebug(_LogPrefix + ": Removing defunct entry for {0}. {1} total subscribers after remove.", observer, _Observers.Count);
                }
            }
        }

        /// <summary>
        /// 发送通知
        /// </summary>
        /// <param name="notification"></param>
        /// <param name="predicate"></param>
        public void Notify(Action<TObserver> notification, IEnumerable<TObserver> targetObserver = null)
        {
            var defunct = default(List<IAddressable>);
            foreach (var observer in _Observers)
            {
                if (targetObserver != null && !targetObserver.Contains(observer.Value)) continue;

                try
                {
                    notification(observer.Value);
                }
                catch (Exception e)
                {
                    _Logger.LogError($"Failed to call method from observer {observer} : {e.Message} \r\n  {e.StackTrace}");
                    defunct ??= new List<IAddressable>();
                    defunct.Add(observer.Key);
                }
            }

            //移除所有有问题的订阅者
            if (defunct != default(List<IAddressable>))
            {
                foreach (var observer in defunct)
                {
                    _Observers.TryRemove(observer, out _);
                    if (_Logger.IsEnabled(LogLevel.Debug))
                        _Logger.LogDebug(_LogPrefix + ": Removing defunct entry for {0}. {1} total subscribers after remove.", observer, _Observers.Count);
                }
            }
        }

        /// <summary>
        /// 获取集合枚举器
        /// </summary>
        /// <returns></returns>
        public IEnumerator<TObserver> GetEnumerator()
        {
            return _Observers.Select(observer => observer.Value).GetEnumerator();
        }

        /// <summary>
        /// 获取集合枚举器
        /// </summary>
        /// <returns></returns>
        IEnumerator IEnumerable.GetEnumerator()
        {
            return this.GetEnumerator();
        }
    }
}
